/**
 * @author zhounan 
 * @date 2024年5月20日
 * @version V1.0
 */
package org.zn.etl.service.impl;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.annotation.PostConstruct;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.zn.etl.job.context.GlobalContextKey;
import org.zn.etl.job.context.JobExecInstanceContext;
import org.zn.etl.job.context.NodeRunningContextManager;
import org.zn.etl.job.entity.EtlJob;
import org.zn.etl.job.entity.JobGroup;
import org.zn.etl.job.entity.po.JobExecInstancePo;
import org.zn.etl.job.handler.ExecStatus;
import org.zn.etl.job.handler.JobHandler;
import org.zn.etl.job.transformer.DataTransformer;
import org.zn.etl.service.EtlExecuteService;
import org.zn.etl.service.EtlPersistService;
import org.zn.etl.service.EtlRunTimePersistService;

import lombok.extern.slf4j.Slf4j;

/**
 * @author zhounan 
 * @date 2024年5月20日
 * @version V1.0
 */
@Service
@Slf4j
public class EtlExecuteServiceImpl implements EtlExecuteService {
	
	@Autowired
	private EtlPersistService  persistService;
	@Autowired
	private Map<String, JobHandler> jobHandlerMap;
	
	@Value(value = "#{${etl.global.context}}")
	private Map<String,Object> globalContext;
	
	@Autowired
	private  Map<String, DataTransformer> dataTransformers;
	
	@Autowired
	private EtlRunTimePersistService runTimePersistService;
	@Autowired
	private NodeRunningContextManager jobContextManager;
	
	private	ExecutorService exec;
	
	@PostConstruct
	private void init() {
		
		Integer poolsize= (Integer) globalContext.get(GlobalContextKey.MAX_POOL_SIZE);
		if(poolsize==null) {
			poolsize=Runtime.getRuntime().availableProcessors()*2;
		}		
				
		exec=Executors.newFixedThreadPool(poolsize);
		
	}


	public synchronized void  addContext(Map<String, Object> map) {
		globalContext.putAll(map);
	}
	
	public Map<String, Object> getGlobalContext() {
		return globalContext;
	}
	
	
	public JobExecInstanceContext executeJob(String jobName ,Map<String, Object> execParam) throws Exception {
		
		EtlJob job = persistService.queryEtlJob(jobName);

		Map<String, Object> execContext = new HashMap<>();

		execContext.putAll(globalContext);
		execContext.putAll(execParam);
		
		String clusterName=(String) execParam.get(GlobalContextKey.CLUSTER_NAME);
		
		JobExecInstanceContext jei = new JobExecInstanceContext();
	
		jei.setJob(job);
		jei.setClusterJobName(clusterName);

		jei.setContextMap(execContext);
		jei.setJobId(job.getId());
		jei.setStartTime(new Date());
		jei.setStatus(ExecStatus.start.getStatus());

		if(CollectionUtils.isNotEmpty(job.getDataMappings()) ) {
			String dataTrans= job.getDataMappings().get(0).getDataTransformer();
			if(StringUtils.isNoneEmpty(dataTrans)) {
				jei.setTrans(dataTransformers.get(dataTrans));
			}
		}
		
		executeJob(jei);

		return jei;
		
	}

	
	private void dojobHandler(JobHandler jobHandler,JobExecInstanceContext ejei) {
		try {
			
			ejei.setExecuteThread(Thread.currentThread());
			jobHandler.executeJob(ejei);
			
			ejei.setEndTime(new Date());
			
			if(! ExecStatus.cannel.getStatus().equals(ejei.getStatus())) {
				ejei.setStatus(ExecStatus.finished.getStatus());				
			}
			
			runTimePersistService.saveExecInstance(ejei);
			
			log.info("job:{} stoped , write data:{}",ejei.getJob().getJobName(),ejei.getTotal());
			
		} catch (Exception e) {
			log.error("executeJob error", e);
			ejei.setEndTime(new Date());
			ejei.setStatus(ExecStatus.error.getStatus());
			runTimePersistService.saveExecInstance(	ejei);
		}finally {
			jobContextManager.renmoveJobExecIns(ejei.getJob().getJobName());
		}
		
	}
	
	

	@Override
	public void executeJob(JobExecInstanceContext jei) throws Exception {
		// TODO Auto-generated method stub
		runTimePersistService.saveExecInstance(jei);
		jobContextManager.putJobExecIns(jei);
		
		String handlerName ="ProtoJobHandler"; 
		
		if(jei.getJob().getHandlerType().equals("proto")) {
			handlerName="ProtoJobHandler";
		}
		else if(jei.getJob().getHandlerType().equals("datax")) {
			handlerName ="DataxJobHandler";
		}else {
			handlerName=jei.getJob().getHandlerType();
		}

		JobHandler jobHandler = jobHandlerMap.get(handlerName);

		if (Boolean.TRUE.equals(jei.getContextMap().get(GlobalContextKey.ASYNS))) {
			exec.execute(() -> {
				dojobHandler(jobHandler, jei);
			});
		}else {
			
			dojobHandler(jobHandler, jei);
			
		}
		
	}

	
	public void executeGroup(String group,Map<String, Object> execParam) {
		
		JobGroup jg=persistService.queryJobGroupBy(group);
		
		execParam.put(GlobalContextKey.ASYNS, false);
		
		Map<String, Object> groupJobParams = (Map<String, Object>) execParam.remove(GlobalContextKey.GJP);
		
		exec.execute(()->{		
			
			for (EtlJob job : jg.getJobList()) {
			
				Map<String, Object> execJobParam=new HashMap<>(execParam) ;
				
				log.info("executeGroup start job:{}",job.getJobName());
				
				if(groupJobParams!=null) {
					
					Map<String, Object> jobParam=  (Map<String, Object>) groupJobParams.get(job.getJobName());
					if(jobParam!=null) {
						execJobParam.putAll(jobParam);
						log.info("executeGroup find jobParam for {} :{}",job.getJobName(),jobParam);
						
					}
					
				}
				
				try {
					
					executeJob(job.getJobName(), execJobParam);
					
				} catch (Exception e) {
					log.error("job:{} execute error", job.getJobName(), e);
					e.printStackTrace();
				}
			
			}
		});

	}


	@Override
	public List<JobExecInstancePo> queryInstanceList(Long jobId, String startTime, String endTime, String status) {
		// TODO Auto-generated method stub
		
		return 	runTimePersistService.queryInstanceList(jobId, startTime, endTime, status);
	}

	

}
